package yeliuhuishi.GroupingComparator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import yeliuhuishi.File.DelAllFile;

import java.io.File;
import java.io.IOException;

/**
 * 1、分组
 * 2、返回每组最大的数据
 * group
 */
public class SecondarySort {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        String input = "src/main/java/yeliuhuishi/GroupingComparator/input";
        String output = "src/main/java/yeliuhuishi/GroupingComparator/output";
        DelAllFile.delAllFile(new File(output));
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, SecondarySort.class.getName());

        job.setMapperClass(SecondarySortMapper.class);
        job.setReducerClass(SecondarySortReducer.class);
        job.setJarByClass(SecondarySort.class);

        job.setMapOutputKeyClass(OrderBean.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setOutputKeyClass(OrderBean.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));

        job.setGroupingComparatorClass(ItemidGroupingComparator.class);

        System.out.println(job.waitForCompletion(true) ? "成功" : "失败");


    }

    static class SecondarySortMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split("\\s");
            OrderBean bean = new OrderBean();
            bean.set(new Text(split[1]), new DoubleWritable(Double.parseDouble(split[2])));
            context.write(bean, NullWritable.get());
        }
    }

    static class SecondarySortReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> {
        //在设置了groupingcomparator以后，这里收到的kv数据 就是：  <1001 87.6>,null  <1001 76.5>,null  ....
        //此时，reduce方法中的参数key就是上述kv组中的第一个kv的key：<1001 87.6>
        //要输出同一个item的所有订单中最大金额的那一个，就只要输出这个key
        @Override
        protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            context.write(key, NullWritable.get());
        }
    }

}
